package org.qq.service.metadata;

import com.github.melin.superior.sql.parser.mysql.MySqlHelper;
import com.google.common.collect.Lists;
import io.github.melin.sqlflow.analyzer.Analysis;
import io.github.melin.sqlflow.analyzer.StatementAnalyzer;
import io.github.melin.sqlflow.metadata.SchemaTable;
import io.github.melin.sqlflow.metadata.SimpleMetadataService;
import io.github.melin.sqlflow.metadata.SqlMetadataExtractor;
import io.github.melin.sqlflow.parser.SqlFlowParser;
import io.github.melin.sqlflow.tree.statement.EmptyStatement;
import io.github.melin.superior.common.relational.Statement;
import io.github.melin.superior.common.relational.create.CreateTable;
import io.github.melin.superior.common.relational.create.CreateTableAsSelect;
import io.github.melin.superior.common.relational.dml.InsertTable;
import io.github.melin.superior.common.relational.dml.MergeTable;
import io.github.melin.superior.parser.flink.FlinkSqlHelper;
import io.github.melin.superior.parser.oracle.OracleSqlHelper;
import io.github.melin.superior.parser.postgre.PostgreSqlHelper;
import io.github.melin.superior.parser.postgre.RedshiftSqlHelper;
import io.github.melin.superior.parser.presto.PrestoSqlHelper;
import io.github.melin.superior.parser.spark.SparkSqlHelper;
import io.github.melin.superior.parser.sqlserver.SqlServerHelper;
import io.github.melin.superior.parser.starrocks.StarRocksHelper;
import io.github.melin.superior.parser.trino.TrinoSqlHelper;
import org.apache.commons.lang3.StringUtils;
import org.qq.enums.DatabaseTypeEnum;
import org.qq.exception.SqlParseException;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

import static java.util.Collections.emptyMap;


@Component
public class MetadataServiceFactory {

    public List<Statement> statements(String databaseType,String sqlStr) {
        if (databaseType == null) {
            throw new SqlParseException("数据库类型不能为空");
        }
        switch (DatabaseTypeEnum.fromString(databaseType.toUpperCase().trim())) {
            case PRESTO:
                return PrestoSqlHelper.parseMultiStatement(sqlStr);
            case TRINO:
                return TrinoSqlHelper.parseMultiStatement(sqlStr);
            case REDSHIFT:
                return RedshiftSqlHelper.parseMultiStatement(sqlStr);
            case FLINK:
                return FlinkSqlHelper.parseMultiStatement(sqlStr);
            case SPARK:
                return SparkSqlHelper.parseMultiStatement(sqlStr);
            case HIVE:
                return SparkSqlHelper.parseMultiStatement(sqlStr);
            case DORIS:
                return StarRocksHelper.parseMultiStatement(sqlStr);
            case STARROCKS:
                return StarRocksHelper.parseMultiStatement(sqlStr);
            case ORACLE:
                return OracleSqlHelper.parseMultiStatement(sqlStr);
            case MYSQL:
                return MySqlHelper.parseMultiStatement(sqlStr);
            case POSTGRES:
                return PostgreSqlHelper.parseMultiStatement(sqlStr);
            case SQLSERVER:
                return SqlServerHelper.parseMultiStatement(sqlStr);
            default:
                throw new SqlParseException("不支持的数据库类型: " + databaseType);
        }
    }


    public List<String> getLineageStrListFromCreate(String databaseType,String sqlStr) throws IOException {
        List<Statement> statements= statements(databaseType,sqlStr);
        SimpleMetadataService metadataService = new SimpleMetadataService("default");
        List<SchemaTable> tables = Lists.newArrayList();
        List<String> sqlList=Lists.newArrayList();
        for (Statement statement : statements) {
            if (statement instanceof CreateTable) {
                CreateTable createTable = (CreateTable) statement;
                List<String> columns = createTable.getColumnRels().stream()
                        .map(column -> column.getColumnName().trim())
                        .collect(Collectors.toList());
                SchemaTable table = new SchemaTable(
                        createTable.getTableId().getCatalogName(),
                        createTable.getTableId().getSchemaName()==null?"default":createTable.getTableId().getSchemaName(),
                        createTable.getTableId().getTableName().trim(),
                        columns,
                        createTable.getPartitionColumnNames()
                );
                tables.add(table);
            }
            if(statement instanceof InsertTable
                    || statement instanceof CreateTableAsSelect
                    ||statement instanceof MergeTable){
                if(StringUtils.isNotBlank(statement.getSql())){
                    sqlList.add(statement.getSql());
                }
            }
        }
        metadataService.addTableMetadata(tables);
        SqlFlowParser SQL_PARSER = new SqlFlowParser();
        // 使用新方法解析多个SQL语句
        List<io.github.melin.sqlflow.tree.statement.Statement> multipleStatement = SQL_PARSER.createStatements(sqlList);
        // 创建StatementAnalyzer实例
        StatementAnalyzer statementAnalyzer = new StatementAnalyzer(
                new Analysis(new EmptyStatement(), emptyMap()), // 创建一个临时Analysis对象
                metadataService,
                SQL_PARSER
        );
        // 使用analyzeMultiple方法处理多个语句
        List<String> lineageStr = statementAnalyzer.analyzeMultiple(multipleStatement, Optional.empty());

        return lineageStr;
    }

    public List<String> getLineageStrList(String databaseType,String sqlStr) throws IOException {
        List<Statement> statements= statements(databaseType,sqlStr);
        SimpleMetadataService metadataService = new SimpleMetadataService("default");
        List<String> sqlList=Lists.newArrayList();
        for (Statement statement : statements) {
            if(statement instanceof InsertTable
                    || statement instanceof CreateTableAsSelect
                    ||statement instanceof MergeTable){
                if(StringUtils.isNotBlank(statement.getSql())){
                    sqlList.add(statement.getSql());
                }
            }
        }
        SqlFlowParser SQL_PARSER = new SqlFlowParser();
        // 使用新方法解析多个SQL语句
        List<io.github.melin.sqlflow.tree.statement.Statement> multipleStatement = SQL_PARSER.createStatements(sqlList);
        for (io.github.melin.sqlflow.tree.statement.Statement statement : multipleStatement) {
            SqlMetadataExtractor extractor = new SqlMetadataExtractor();
            // 清理上下文，确保开始处理前状态干净
            extractor.clearContext();
            extractor.process(statement);
            // 3. 创建 SimpleMetadataService 并添加元数据
            // 所有表和字段信息
            for (SchemaTable table : extractor.getTables()) {
                metadataService.addTableMetadata(table);
            }
            //清理线程本地变量，避免内存泄漏
            extractor.removeContext();
        }
        // 创建StatementAnalyzer实例
        StatementAnalyzer statementAnalyzer = new StatementAnalyzer(
                new Analysis(new EmptyStatement(), emptyMap()), // 创建一个临时Analysis对象
                metadataService,
                SQL_PARSER
        );
        // 使用analyzeMultiple方法处理多个语句
        List<String> lineageStr = statementAnalyzer.analyzeMultiple(multipleStatement, Optional.empty());

        return lineageStr;
    }

    public List<String> getLineageStrList(String databaseType,String isCreateTabl,String sqlStr) throws IOException {
        List<Statement> statements= statements(databaseType,sqlStr);
        SimpleMetadataService metadataService = new SimpleMetadataService("default");
        List<String> sqlList=Lists.newArrayList();
        List<String> tables = Lists.newArrayList();
        for (Statement statement : statements) {
            if ("true".equals(isCreateTabl) && statement instanceof CreateTable) {
                CreateTable createTable = (CreateTable) statement;
                List<String> columns = createTable.getColumnRels().stream()
                        .map(column -> column.getColumnName().trim())
                        .collect(Collectors.toList());
                SchemaTable table = new SchemaTable(
                        createTable.getTableId().getCatalogName(),
                        createTable.getTableId().getSchemaName()==null?"default":createTable.getTableId().getSchemaName(),
                        createTable.getTableId().getTableName().trim(),
                        columns,
                        createTable.getPartitionColumnNames()
                );
                tables.add(table.getFullTableName());
                metadataService.addTableMetadata(table);
            }
            if(statement instanceof InsertTable
                    || statement instanceof CreateTableAsSelect
                    ||statement instanceof MergeTable){
                if(StringUtils.isNotBlank(statement.getSql())){
                    sqlList.add(statement.getSql());
                }
            }
        }
        SqlFlowParser SQL_PARSER = new SqlFlowParser();
        // 使用新方法解析多个SQL语句
        List<io.github.melin.sqlflow.tree.statement.Statement> multipleStatement = SQL_PARSER.createStatements(sqlList);
        for (io.github.melin.sqlflow.tree.statement.Statement statement : multipleStatement) {
            SqlMetadataExtractor extractor = new SqlMetadataExtractor();
            // 清理上下文，确保开始处理前状态干净
            extractor.clearContext();
            extractor.process(statement);
            // 3. 创建 SimpleMetadataService 并添加元数据
            // 所有表和字段信息
            for (SchemaTable table : extractor.getTables()) {
                if(!tables.contains(table.getFullTableName())){
                    metadataService.addTableMetadata(table);
                }
            }
            //清理线程本地变量，避免内存泄漏
            extractor.removeContext();
        }
        // 创建StatementAnalyzer实例
        StatementAnalyzer statementAnalyzer = new StatementAnalyzer(
                new Analysis(new EmptyStatement(), emptyMap()), // 创建一个临时Analysis对象
                metadataService,
                SQL_PARSER
        );
        // 使用analyzeMultiple方法处理多个语句
        List<String> lineageStr = statementAnalyzer.analyzeMultiple(multipleStatement, Optional.empty());

        return lineageStr;
    }

}